//package webflux.reactive.stream;
//
//import lombok.extern.slf4j.Slf4j;
//import org.junit.Test;
//
//import java.util.concurrent.Flow;
//import java.util.concurrent.SubmissionPublisher;
//
///**
// * @program JavaBase
// * @description:
// * @author: zhanglu
// * @create: 2021-05-18 16:44:00
// */
//@Slf4j
//public class ReactiveStreamDemo {
//
//    @Test
//    public void test01() throws InterruptedException {
//        //创建一个发布者
//        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
//        //创建一个订阅者
//        Flow.Subscriber subscriber = new Flow.Subscriber<>() {
//
//            Flow.Subscription subscription;
//
//            @Override
//            public void onSubscribe(Flow.Subscription subscription) {
//                log.info("建立订阅关系");
//                this.subscription = subscription;
//                //第一次建立关系时需要发送个消息
//                subscription.request(1);
//            }
//
//            @Override
//            public void onNext(Object item) {
//                log.info("接收数据:" + item);
//                //背压
//                subscription.request(10);
//            }
//
//            @Override
//            public void onError(Throwable throwable) {
//                log.info("异常:" + throwable.getMessage());
//            }
//
//            @Override
//            public void onComplete() {
//                log.info("接受完成～");
//            }
//        };
//        //建立订阅关系
//        publisher.subscribe(subscriber);
//        //发送数据
//        for (int i = 0; i < 1; i++) {
//            publisher.submit("hello lu~");
//        }
//        Thread.currentThread().join();
//    }
//
//    @Test
//    public void test02() throws InterruptedException {
//        //创建一个发布者
//        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
//        //创建一个处理器
//        ReactiveProcessor processor = new ReactiveProcessor();
//        //发布者与处理器建立关系
//        publisher.subscribe(processor);
//        //创建一个订阅者
//        Flow.Subscriber<String> subscriber = new Flow.Subscriber<>() {
//            Flow.Subscription subscription;
//            @Override
//            public void onSubscribe(Flow.Subscription subscription) {
//                log.info("建立订阅关系");
//                this.subscription = subscription;
//                //第一次建立关系时需要发送个消息
//                subscription.request(1);
//            }
//
//            @Override
//            public void onNext(String item) {
//                log.info("接收数据:" + item);
//                //背压
//                subscription.request(10);
//            }
//
//            @Override
//            public void onError(Throwable throwable) {
//                log.info("异常:" + throwable.getMessage());
//            }
//
//            @Override
//            public void onComplete() {
//                log.info("接受完成～");
//            }
//        };
//        //建立订阅关系
//        publisher.subscribe(subscriber);
//        //发送数据
//        for (int i = 0; i < 4; i++) {
//            publisher.submit("hello lu~");
//        }
//        Thread.currentThread().join();
//    }
//
//}
